package org.atguigu.gmall.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.SQLUtil;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dwd.db
 * Author: ZengHaiFeng
 * CreateTime:2023/4/25 18:34:12
 * Description: Todo
 * Version：1.0
 */
public class Dwd_04_DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_04_DwdTradeOrderDetail().init(30004,2,"Dwd_04_DwdTradeOrderDetail");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        // step 1. read data from ods_db
        readOdsDb(tbe,"Dwd_04_DwdTradeOrderDetail");
        // step 2. filter order detail data
        Table orderDetail = tbe.sqlQuery(" select " +
                " `data`['id'] id, " +
                " `data`['order_id'] order_id, " +
                " `data`['sku_id'] sku_id, " +
                " `data`['sku_name'] sku_name, " +
                " `data`['create_time'] create_time, " +
                " `data`['sku_num'] sku_num, " +
                " cast(cast(`data`['sku_num'] as decimal(16,2)) * cast(`data`['order_price'] as decimal(16,2)) as string) " +
                "       split_original_amount, " +
                " `data`['split_total_amount'] split_total_amount, " +
                " `data`['split_activity_amount'] split_activity_amount, " +
                " `data`['split_coupon_amount'] split_coupon_amount, " +
                " `ts` " +
                " from ods_db where " +
                " `database`='gmall2023' and `table`='order_detail' and `type`='insert' ");
        tbe.createTemporaryView("order_detail",orderDetail);

        // step 3. filter order info data
        Table orderInfo = tbe.sqlQuery("select" +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['province_id'] province_id " +
                " from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='order_info' and " +
                " `type`='insert' ");
        tbe.createTemporaryView("order_info",orderInfo);

        // step 4. filter order detail activity
        Table orderDetailActivity = tbe.sqlQuery("select " +
                " `data`['order_detail_id'] order_detail_id, " +
                " `data`['activity_id'] activity_id, " +
                " `data`['activity_rule_id'] activity_rule_id " +
                " from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='order_detail_activity' and " +
                " `type`='insert' ");
        tbe.createTemporaryView("order_detail_activity",orderDetailActivity);

        // step 5. filter order detail coupon
        Table orderDetailCoupon = tbe.sqlQuery("select " +
                " `data`['order_detail_id'] order_detail_id, " +
                " `data`['coupon_id'] coupon_id " +
                " from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='order_detail_coupon' and " +
                " `type`='insert' ");
        tbe.createTemporaryView("order_detail_coupon",orderDetailCoupon);

        // step 6. there are four tables do join
        Table result = tbe.sqlQuery("select " +
                " od.id, " +
                " od.order_id, " +
                " oi.user_id, " +
                " od.sku_id, " +
                " od.sku_name, " +
                " oi.province_id, " +
                " oa.activity_id, " +
                " oa.activity_rule_id, " +
                " oc.coupon_id, " +
                " date_format(od.create_time,'yyyy-MM-dd') as date_id, " +
                " od.create_time, " +
                " od.sku_num, " +
                " od.split_original_amount, " +
                " od.split_activity_amount, " +
                " od.split_coupon_amount, " +
                " od.split_total_amount, " +
                " od.ts " +
                " from order_detail as od " +
                " join order_info as oi " +
                " on od.order_id=oi.id " +
                " left join order_detail_activity as oa on od.id=oa.order_detail_id " +
                " left join order_detail_coupon as oc on od.id=oc.order_detail_id ");

        // step 7. put the final result into a kafka topic
        tbe.executeSql("create table dwd_trade_order_detail(" +
                " id string, " +
                " order_id string, " +
                " user_id string, " +
                " sku_id string, " +
                " sku_name string, " +
                " province_id string, " +
                " activity_id string, " +
                " activity_rule_id string, " +
                " coupon_id string, " +
                " date_id string, " +
                " create_time string, " +
                " sku_num string, " +
                " split_original_amount string, " +
                " split_activity_amount string, " +
                " split_coupon_amount string, " +
                " split_total_amount string, " +
                " ts bigint, " +
                " primary key(id) not enforced" +
                ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
        result.executeInsert("dwd_trade_order_detail");
    }
}
